package com.ams.io.network;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.io.network.connection.ConnectionListener;
import com.ams.io.network.connection.SocketConnector;

public class SocketAcceptor extends NetworkHandler {
    final private Logger logger = LoggerFactory.getLogger(SocketAcceptor.class);

    private SocketAddress listenEndpoint;
    private SocketProperties socketProperties = null;
    private ServerSocketChannel serverChannel;
    private Selector selector;
    private int dispatcherSize = 1;

    private ArrayList<Dispatcher> dispatchers = new ArrayList<Dispatcher>();
    private int nextDispatcher = 0;
    private ConnectionListener listener;

    public SocketAcceptor(SocketAddress host) throws IOException {
        listenEndpoint = host;
        createDispatcher();
    }

    public SocketAcceptor(SocketAddress host, int dispatcherSize)
            throws IOException {
        listenEndpoint = host;
        this.dispatcherSize = dispatcherSize;
        createDispatcher();
    }

    private void createDispatcher() throws IOException {
        for (int i = 0; i < dispatcherSize; i++) {
            Dispatcher dispatcher = new Dispatcher();
            dispatchers.add(dispatcher);
        }
    }

    private void openChannel() throws IOException {
        serverChannel = ServerSocketChannel.open();
        serverChannel.socket().bind(listenEndpoint);
        serverChannel.configureBlocking(false);
        selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    private void closeChannel() {
        try {
            serverChannel.close();
            selector.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void run() {
        int selectedKeys = 0;
        while (isRunning()) {
            try {
                selectedKeys = selector.select();
            } catch (Exception e) {
                if (selector.isOpen()) {
                    continue;
                } else {
                    try {
                        selector.selectNow();
                        selector = Selector.open();
                        serverChannel
                                .register(selector, SelectionKey.OP_ACCEPT);
                    } catch (Exception e1) {
                    }
                }
            }
            if (selectedKeys == 0) {
                continue;
            }

            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();
                it.remove();

                if (!key.isValid()) {
                    continue;
                }

                try {
                    ServerSocketChannel serverChannel = (ServerSocketChannel) key
                            .channel();
                    SocketChannel channel = serverChannel.accept();

                    //if (socketProperties != null) {
                    //     socketProperties.setSocketProperties(channel.socket());
                    //}

                    if (dispatchers != null) {
                        Dispatcher dispatcher = dispatchers
                                .get(nextDispatcher++);

                        SocketConnector connector = new SocketConnector(channel);
                        if (listener != null) {
                            connector.addListener(listener);
                        }

                        dispatcher.addChannelToRegister(new ChannelInterestOps(
                                channel, SelectionKey.OP_READ, connector));
                        if (nextDispatcher >= dispatchers.size()) {
                            nextDispatcher = 0;
                        }
                    }
                } catch (Exception e) {
                    key.cancel();
                }
            }
        }
    }

    public void setSocketProperties(SocketProperties socketProperties) {
        this.socketProperties = socketProperties;
    }

    public SocketAddress getListenEndpoint() {
        return this.listenEndpoint;
    }

    public void start(String name) {
        for (int i = 0; i < dispatchers.size(); i++) {
            Dispatcher dispatcher = dispatchers.get(i);
            dispatcher.start("socket dispatcher");
        }

        try {
            openChannel();
        } catch (IOException e) {
            e.printStackTrace();
        }
        super.start("socket acceptor");
    }

    public void stop() {
        for (int i = 0; i < dispatchers.size(); i++) {
            dispatchers.get(i).stop();
        }

        closeChannel();
        super.stop();
    }

    public void setConnectionListener(ConnectionListener listener) {
        this.listener = listener;
    }

}
